package com.aiobc.mq;

import com.aiobc.mq_batch.MessageListSplitter;
import com.aiobc.mq_transaction.TransactionProducer;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;

@Slf4j
@Component
public class MQProducerService {

    @Value("${rocketmq.producer.send-message-timeout}")
    private Integer messageTimeOut;

    // 建议正常规模项目统一用一个TOPIC
    private static final String topic = "xc_topic";

    // 直接注入使用，用于发送消息到broker服务器
    @Resource
    private RocketMQTemplate rocketMQTemplate;
    @Resource
    private TransactionProducer producer;
    @Resource
    private MQProducerConfigure producerConfigure;
    /**
     * 普通发送
     */
    public void send(String msg) {
        rocketMQTemplate.convertAndSend(topic + ":tag1", msg);
//        rocketMQTemplate.send(topic + ":tag1", MessageBuilder.withPayload(user).build()); // 等价于上面一行
    }

    /**
     * 发送同步消息（阻塞当前线程，等待broker响应发送结果，这样不太容易丢失消息）
     * （msgBody也可以是对象，sendResult为返回的发送结果）
     */
    public SendResult sendSynMsg(String msgBody) {
        SendResult sendResult = rocketMQTemplate.syncSend(topic,MessageBuilder.withPayload(msgBody).build());
        log.info("【sendMsg】sendResult={}", JSON.toJSONString(sendResult));
        return sendResult;
    }

    /**
     * 发送异步消息（通过线程池执行发送到broker的消息任务，执行完后回调：在SendCallback中可处理相关成功失败时的逻辑）
     * （适合对响应时间敏感的业务场景）
     */
    public void sendAsyncMsg(String msgBody) {
        rocketMQTemplate.asyncSend(topic, MessageBuilder.withPayload(msgBody).build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("=========发送成功");
                // 处理消息发送成功逻辑
            }
            @Override
            public void onException(Throwable throwable) {
                System.out.println("=========发送异常"+throwable);
                // 处理消息发送异常逻辑
            }
        });
    }

    /**
     * 发送延时消息（上面的发送同步消息，delayLevel的值就为0，因为不延时）
     * 在start版本中 延时消息一共分为18个等级分别为：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     */
    public void sendDelayMsg(String msgBody, int delayLevel){
        rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(msgBody).build(), messageTimeOut, delayLevel);

    }

    /**
     * 发送单向消息（只负责发送消息，不等待应答，不关心发送结果，如日志）
     */
    public void sendOneWayMsg(String msgBody) {
        rocketMQTemplate.sendOneWay(topic, MessageBuilder.withPayload(msgBody).build());
    }

    /**
     * 发送带tag的消息，直接在topic后面加上":tag"
     */
    public SendResult sendTagMsg(String msgBody) {
        return rocketMQTemplate.syncSend(topic + ":tag2", MessageBuilder.withPayload(msgBody).build());
    }
    /**
     * 发送事务消息
     * 事务消息使用上的限制
     * 1:事务消息不支持延时消息和批量消息。
     * 2:为了避免单个消息被检查太多次而导致半队列消息累积，我们默认将单个消息的检查次数限制为 15 次，但是用户可以通过 Broker 配置文件的 transactionCheckMax参数来修改此限制。如果已经检查某条消息超过 N 次的话（ N = transactionCheckMax ） 则 Broker 将丢弃此消息，并在默认情况下同时打印错误日志。用户可以通过重写 AbstractTransactionalMessageCheckListener 类来修改这个行为。
     * 3:事务消息将在 Broker 配置文件中的参数 transactionTimeout 这样的特定时间长度之后被检查。当发送事务消息时，用户还可以通过设置用户属性 CHECK_IMMUNITY_TIME_IN_SECONDS 来改变这个限制，该参数优先于 transactionTimeout 参数。
     * 4:事务性消息可能不止一次被检查或消费。
     * 5:提交给用户的目标主题消息可能会失败，目前这依日志的记录而定。它的高可用性通过 RocketMQ 本身的高可用性机制来保证，如果希望确保事务消息不丢失、并且事务完整性得到保证，建议使用同步的双重写入机制。
     * 6:事务消息的生产者 ID 不能与其他类型消息的生产者 ID 共享。与其他类型的消息不同，事务消息允许反向查询、MQ服务器能通过它们的生产者 ID 查询到消费者。
     */
    public void sendTransactionMsg(String msgBody) throws InterruptedException, MQClientException {

        //发送事务消息
//        TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(
//                "send_message_to_xc_topic",
//                MessageBuilder.withPayload(JSONUtil.toJsonStr(msgBody)).build(),
//                null);
        producer.sendTransactionMsg(msgBody);
    }

    /**
     * 批量发送消息
     * @param msgBody
     * @throws MQClientException
     */
    public void batchSendMsg(String msgBody) throws MQClientException, InterruptedException {
        DefaultMQProducer mqProducer = producerConfigure.defaultProducer();
        // 定义要发送的消息集合
        List<org.apache.rocketmq.common.message.Message> messages = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            byte[] body = (msgBody + i).getBytes();
            System.out.println("==="+body);
            org.apache.rocketmq.common.message.Message msg = new org.apache.rocketmq.common.message.Message(topic, "*", body);
            messages.add(msg);
        }

        // 定义消息列表分割器，将消息列表分割为多个不超出4M大小的小列表
        MessageListSplitter splitter = new MessageListSplitter(messages);
        while (splitter.hasNext()) {
            try {
                List<org.apache.rocketmq.common.message.Message> listItem = splitter.next();

                mqProducer.send(listItem);

            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        //线程睡1秒
//        TimeUnit.SECONDS.sleep(1);
        //6.关闭生产者producer
//        mqProducer.shutdown();
    }
}

